package com.example.mq.mqserver.core;

import com.example.mq.common.Consumer;
import com.example.mq.common.ConsumerEnv;
import com.example.mq.common.MqException;
import com.example.mq.mqserver.VirtualHost;
import lombok.extern.slf4j.Slf4j;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * 通过这个类来实现消费消息的核心逻辑
 */
@Slf4j
public class ConsumerManager {
    //存储上层的VirtualHost的对象引用，用来操作数据
    private VirtualHost parent;
    //存放令牌的队列
    private BlockingQueue<String> tokens = new LinkedBlockingQueue<>();
    //扫描整体队列的线程
    private Thread thread;
    //用来存放当前所有队列中有哪些有消费者（为后续队列实现非独占做准备）
    private ArrayList<ConsumerEnv> consumerLists = new ArrayList<>();
    //方便采用轮询（第一次消费过后需要重新排队进行消费）
    private AtomicInteger seqNumber = new AtomicInteger(0);
    //存放所有有消息的集合
    private ArrayList<String> hasMessageConsumer = new ArrayList<>();

    public void putMassageQueue(String queueName) {
        hasMessageConsumer.add(queueName);
    }
    public ConsumerManager(VirtualHost virtualHost) {
        parent = virtualHost;
        thread = new Thread(() -> {
            while(true) {
                try {
                    //1.拿到令牌（队列名）
                    String queueName = tokens.take();
                    //2.根据令牌找到队列
                    MSGQueue queue = parent.getMemoryDataManage().getQueue(queueName);
                    if(queue == null) {
                        throw new MqException("当前队列不存在，queueName: " + queueName);
                    }
                    //3.消费消息(有可能产生线程安全问题所以需要对当前队列加锁)
                    synchronized(queue) {
                        try {
                            consumeMessage(queue);
                        }catch (MqException e) {
                            e.printStackTrace();
                            log.info("[consumerDaemon] 消费信息失败");
                        }
                    }
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        });
        //设置为后台线程
        thread.setDaemon(true);
        thread.start();
    }

    public void notifyConsumer(String queueName) throws InterruptedException {
        tokens.put(queueName);
    }

    public void addConsumer(String consumerTag, String queueName, boolean autoAck, Consumer consumer) {
        try {
            //1.根据队列名找到队列
            MSGQueue queue = parent.getMemoryDataManage().getQueue(queueName);
            if(queue == null) {
                throw new MqException("[consumerManager] 当前队列不存在,queueName: " + queueName);
            }
            //2.构造消费者
            ConsumerEnv consumerEnv = new ConsumerEnv(consumerTag,queueName,autoAck,consumer);
            //3.添加消费者
            synchronized (queue) {
                parent.getMemoryDataManage().addConsumer(consumerTag,consumerEnv);
                queue.addConsumerEnv(consumerEnv);
                consumerLists.add(consumerEnv);
                tokens.addAll(hasMessageConsumer);
                //如果当前队列中已经有了消息就直接消费掉
                int totalMessages = parent.getMemoryDataManage().getCounts(queueName);
                while(totalMessages > 0) {
                    //每一次循环就消费掉一个消息
                    consumeMessage(queue);
                    totalMessages--;
                }
            }
            log.info("[consumerManager] 添加消费者到队列中成功");
        }catch (Exception e) {
            e.printStackTrace();
            log.info("[consumerManager] 添加消费者到队列中失败,queueName: " + queueName + " ,consumerTag: " + consumerTag);
        }
    }

    private void consumeMessage(MSGQueue queue) throws InterruptedException {
        //1.按照轮询的方式，查询出一个消费者
        ConsumerEnv luckier = null;
        log.info("queueName: " + queue.getName() + ",exclusive: " +  queue.isExclusive());

        if(queue.isExclusive()) {
            //独占
            luckier = queue.chooseConsumerEnvExclusive();
            if (luckier == null) {
                throw new MqException("[consumerManager] 当前队列中没有消费者,queueName: " + queue.getName());
            }
            log.info("独占 queueName: " + luckier.getQueueName() + ",exclusive: " +  queue.isExclusive());
        }else {
            //非独占
            luckier = chooseConsumerEnvNotExclusive();
            if (luckier == null) {
                throw new MqException("[consumerManager] 当前所有队列中没有消费者,queueName: " + queue.getName());
            }
            log.info("非独占 queueName: " + luckier.getQueueName() + ",exclusive: " +  queue.isExclusive());
        }

        //2.从当前有消息的队列中取出一个消息取出一个消息(实际内存还存在)
        Message message = parent.getMemoryDataManage().pollMessage(queue.getName());
        if(message == null) {
            throw new MqException("[consumerManager] 当前队列中没有消息,queueName " + queue.getName());
        }

        ConsumerEnv finalLuckier = luckier;
        //3.消费消息
        try {
            //将未确认的消息存储在内存中
            parent.getMemoryDataManage().addMessageWaitAck(queue.getName(), message);
            //执行回调函数，消费消息
            try {
                finalLuckier.getConsumer().handleDelivery(finalLuckier.getConsumerTag(), message.getBasicProperties(), message.getBody());
            }catch (Exception e) {
                log.info("[consumeMessage] 重新添加消息....");
                parent.getMemoryDataManage().receiveMessage(queue.getName(), message);
                throw new MqException("[consumeMessage] 执行回调失败");
            }
            //这里有两种确认方式：
            // 手动确认：需要手动basicAck才能彻底将消息进行消费
            // 自动确认：无需确认消息一经调用就被消费掉
            if(finalLuckier.isAutoAck()) {
                //删除在硬盘上存储的消息
                if(message.getBasicProperties().getDeliveryMode() == 2) {
                    parent.getDiskDataManage().deleteMessage(queue, message);
                }
                //删除在内存中的存储的消息（未确认）
                parent.getMemoryDataManage().removeMessageWaitAck(queue.getName(), message.getBasicProperties().getMessageId());
                //删除在内存中的存储的消息（实际存储的），删除总的存储message地方中的message
                parent.getMemoryDataManage().removeMessage(message.getBasicProperties().getMessageId());
            }
            log.info("[ConsumerManager] 消费消息成功");
        }catch (Exception e) {
            e.printStackTrace();
            log.info("[ConsumerManager] 消费消息失败");
        }
    }

    private ConsumerEnv chooseConsumerEnvNotExclusive() {
        if(consumerLists.size() == 0) {
            return null;
        }
        int index = seqNumber.get() % consumerLists.size();
        seqNumber.incrementAndGet();
        return consumerLists.get(index);
    }
}
